package com.zyh.flink.day04;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

public class CouponBroadcastStateJob1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Coupon> couponStream = environment.socketTextStream("hadoop10", 9991)
                .map(new MapFunction<String, Coupon>() {
                    @Override
                    public Coupon map(String s) throws Exception {
                        String[] ss = s.split("\\s+");
                        return new Coupon(ss[0],Double.valueOf(ss[1]),Double.valueOf(ss[2]));
                    }
                });

        SingleOutputStreamOperator<Order> orderStream = environment.socketTextStream("hadoop10", 9992)
                .map(new MapFunction<String, Order>() {
                    @Override
                    public Order map(String s) throws Exception {
                        String[] ss = s.split("\\s+");
                        return new Order(ss[0], ss[1], Double.valueOf(ss[2]));
                    }
                });

        KeyedStream<Order, Tuple2<String, String>> keyedStream = orderStream.keyBy(order -> Tuple2.of(order.getUserNo(), order.getCategory()),Types.TUPLE(Types.STRING,Types.STRING));

        MapStateDescriptor<String, Coupon> msd = new MapStateDescriptor<>("coupons", Types.STRING, Types.POJO(Coupon.class));
        BroadcastStream<Coupon> couponBroadcastStream = couponStream.broadcast(msd);

        BroadcastConnectedStream<Order, Coupon> connectedStream = keyedStream.connect(couponBroadcastStream);

        SingleOutputStreamOperator<String> result = connectedStream.process(new MyKeyedBroadcastProcessFunction1(msd));
        result.print();

        environment.execute("KeyedStream");
    }
}
/*
* KS:高速流上的分组依据类型 Tuple2<String,String>
* IN1:高速流上的元素类型Order
* IN2:广播流的数据类型；低吞吐量流的数据类型
* OUT:输出数据类型
* */
class MyKeyedBroadcastProcessFunction1  extends KeyedBroadcastProcessFunction<String,Order,Coupon,String>{
    private MapStateDescriptor<String,Coupon> msd;
    private ReducingState<Double> reducingState;

    MyKeyedBroadcastProcessFunction1(MapStateDescriptor<String,Coupon> msd){
        this.msd = msd;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        RuntimeContext ctx = getRuntimeContext();
        ReducingStateDescriptor<Double> rsd = new ReducingStateDescriptor<>("total", (v1, v2) -> v1 + v2, Types.DOUBLE);
        this.reducingState = ctx.getReducingState(rsd);
    }

    @Override
    public void processElement(Order order, ReadOnlyContext readOnlyContext, Collector<String> collector) throws Exception {
        this.reducingState.add(order.getMoney());
        //获取广播状态中的优惠券
        ReadOnlyBroadcastState<String, Coupon> broadcastState = readOnlyContext.getBroadcastState(this.msd);
        Coupon coupon = broadcastState.get(order.getCategory());
        if (coupon!=null){
            //该类别下有优惠券，判断是否达标
            //获取用户在当前类别的累计消费 total
            Double total = this.reducingState.get();

            if (total>=coupon.getFull()){
                collector.collect(order.getUserNo()+"在类别："+order.getMoney()+"已经消费达标，满"+coupon.getFull()+"减"+coupon.getRebate());
                this.reducingState.clear();
                this.reducingState.add(total-coupon.getFull());
            }else {
                Double a = coupon.getFull()-total;
                collector.collect("还差"+a+"元达标");
            }
        }else {
            collector.collect(order.getCategory()+"这个类别下没有优惠券");
        }
    }

    @Override
    public void processBroadcastElement(Coupon coupon, Context context, Collector<String> collector) throws Exception {
        //获取广播状态
        BroadcastState<String, Coupon> broadcastState = context.getBroadcastState(msd);
        //将低速流上的优惠券放入到广播状态中
        broadcastState.put(coupon.getCategory(),coupon);
    }
}